package com.neusoft.databus.policy.expression.impl;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.jexl3.JexlContext;
import org.apache.commons.jexl3.MapContext;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.neusoft.bizcore.common.cache.CacheService;
import com.neusoft.bizcore.web.exception.CustomRuntimeException;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.bean.MetricBean;
import com.neusoft.databus.common.bean.ResourceBean;
import com.neusoft.databus.common.constant.DatabusErrorCode;
import com.neusoft.databus.common.utils.DefaultThreadFactory;
import com.neusoft.databus.policy.bean.ComplexPolicyExtendBean;
import com.neusoft.databus.policy.bean.EventBean;
import com.neusoft.databus.policy.config.DatabusPolicyProperties;
import com.neusoft.databus.policy.constants.DatabusPolicyConstant;
import com.neusoft.databus.policy.enums.PolicyExpType;
import com.neusoft.databus.policy.enums.PolicyType;
import com.neusoft.databus.policy.model.Expression;
import com.neusoft.databus.policy.model.Policy;
import com.neusoft.databus.policy.model.PolicyRefResource;
import com.neusoft.databus.policy.repository.PolicyRefResourceRepository;
import com.neusoft.databus.policy.repository.PolicyRepository;
import com.neusoft.databus.policy.utils.ExpressionCalculator;
import com.neusoft.databus.policy.utils.PolicyUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * 复杂告警解析类
 *
 * @author naxiang
 */
@Slf4j
@Component("complexPolicyExpressionHandler")
public class ComplexPolicyExpressionHandlerImpl extends PolicyExpressionHandler {

    @Autowired
    private DatabusPolicyProperties policyProperties;
    @Autowired
    private PolicyRepository policyRepository;
    @Autowired
    private PolicyRefResourceRepository policyRefResourceRepository;
    @Autowired
    private CacheService cacheService;

    private static ThreadPoolExecutor dataExecutor;

    @Override
    public void execute(final CommandBean commandBean) {
        if (ComplexPolicyExpressionHandlerImpl.dataExecutor == null) {
            synchronized (PolicyExpressionHandlerImpl.class) {
                final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);
                ComplexPolicyExpressionHandlerImpl.dataExecutor = new ThreadPoolExecutor(
                        Integer.valueOf(this.policyProperties.getComplexExpressionHandlerThreadSize()),
                        Integer.valueOf(this.policyProperties.getComplexExpressionHandlerThreadMaxSize()),
                        this.policyProperties.getComplexExpressionHandlerKeepAliveTime(), TimeUnit.MICROSECONDS, queue,
                        new DefaultThreadFactory("cx-policy-ex"),
                        new ThreadPoolExecutor.CallerRunsPolicy());
            }
        }
        DataHandlerRunnable runnable = new DataHandlerRunnable(commandBean);
        ComplexPolicyExpressionHandlerImpl.dataExecutor.execute(runnable);
        runnable = null;
    }

    public class DataHandlerRunnable implements Runnable {
        private CommandBean commandBean;

        public DataHandlerRunnable(final CommandBean commandBean) {
            this.commandBean = commandBean;
        }

        @Override
        public void run() {
            final String traceId = this.commandBean.generateTraceId();
            MDC.put("trace", traceId);
            // 获取该资源相关策略
            try {
                // 错误码判断
                final String errorCode = this.commandBean.getErrorCode();
                for (final ResourceBean resourceBean : this.commandBean.getResourceBean()) {
                    final Date time = this.commandBean.getCollectEndTime();
                    if (StringUtils.isBlank(errorCode)) {
                        ComplexPolicyExpressionHandlerImpl.this.sendFixSysEvent(resourceBean, time);
                    }
                    final String resourceCode = resourceBean.getCode();
                    final List<PolicyRefResource> policyRefResources =
                            ComplexPolicyExpressionHandlerImpl.this.policyRefResourceRepository
                                    .findByResourceCode(resourceCode);
                    final List<Policy> policyList = Lists.newArrayList();
                    policyRefResources.forEach(policyRefResource -> {
                        final Policy policy = policyRefResource.getPolicy();
                        if (!policy.isDisabled()) {
                            if (PolicyType.POLICY_ALARM.equals(policy.getPolicyType())) {
                                policyList.add(policy);
                            }
                        }
                    });

                    if ((policyList == null) || policyList.isEmpty()) {
                        ComplexPolicyExpressionHandlerImpl.log.debug("资源{}无相关复杂策略", resourceCode);
                    } else {
                        ComplexPolicyExpressionHandlerImpl.log.debug("资源 {}相关复杂策略个数为[{}]", resourceCode,
                                policyList.size());
                        policyList.forEach(policy -> {
                            ComplexPolicyExpressionHandlerImpl.this.pidHandler(resourceBean, policy, time);
                        });
                    }
                }

            } catch (final Exception e) {
                ComplexPolicyExpressionHandlerImpl.log.error("复杂策略告警解析异常：{}", e);
            } finally {
                this.commandBean = null;
                MDC.clear();
            }
        }
    }

    /**
     * 系统告警解除
     *
     * @param resourceBean
     * @param time
     */
    private void sendFixSysEvent(final ResourceBean resourceBean, final Date time) {
        final EventBean eventBean = new EventBean();
        eventBean.setExpressionResult(false);
        eventBean.setResourceBean(resourceBean);
        eventBean.setExpressionTrueSquence(null);
        final List<PolicyType> types = Lists.newArrayList();
        types.add(PolicyType.DEVICE_FAULT);
        types.add(PolicyType.GATEWAY_DISCONNECT);
        types.add(PolicyType.SYSTEM);
        List<Policy> policies = Lists.newArrayList();
        policies = this.policyRepository.findByPolicyTypeIn(types);
        eventBean.setPolicies(policies);
        eventBean.setTime(time);
        this.sendResult(eventBean);
    }

    /**
     * 指标公式解析类
     *
     * @param commandBean
     * @param policy
     * @param time
     */
    private void pidHandler(ResourceBean resourceBean, final Policy policy, final Date time) {
        final String cacheName = DatabusPolicyConstant.POLICY_EXPRESSION_CACHE_NAME;
        final List<Expression> expList = policy.getExpressions();
        final Long policyId = policy.getId();
        EventBean eventBean = null;
        JexlContext jexlContext = new MapContext();
        try {
            eventBean = new EventBean();
            if (expList.get(0).getType() != PolicyExpType.COMPLEX) {
                ComplexPolicyExpressionHandlerImpl.log.warn("该策略不是复杂策略！，{}", policyId);
                return;
            }
            boolean result = false;
            List<ComplexPolicyExtendBean> complexPolicyExtendBeans = Lists.newArrayList();
            complexPolicyExtendBeans = JsonUtils.jsonToPojoList(policy.getExtendProperty(),
                    new TypeReference<List<ComplexPolicyExtendBean>>() {
                    });
            final String complexPolicyKey = PolicyUtil.getComplexPolicyKey(resourceBean.getCode(), policy);
            out: for (final ComplexPolicyExtendBean extendBean : complexPolicyExtendBeans) {
                final Integer dataType = extendBean.getDataType();
                in: switch (dataType) {
                case 1:
                    result = this.getRealtime(complexPolicyKey, resourceBean.getMetrics(), extendBean);
                    ComplexPolicyExpressionHandlerImpl.log.info("复杂策略：{}，实时数据处理", policyId);
                    if (!result) {
                        break out;
                    }
                    break in;
                case 2:
                    result = this.getHistory(complexPolicyKey, resourceBean.getMetrics(), extendBean);
                    ComplexPolicyExpressionHandlerImpl.log.info("复杂策略：{}，历史数据处理", policyId);
                    if (!result) {
                        break out;
                    }
                    break in;
                }
            }
            if (!result) {
                return;
            }
            jexlContext = (JexlContext) this.cacheService.get(cacheName, complexPolicyKey);
            if (!this.iSDataIntegrity(complexPolicyExtendBeans, jexlContext)) {
                return;
            }
            final Map<String, Object> exp = JsonUtils.jsonToMap(expList.get(0).getExpressionInfo());
            result = ExpressionCalculator.getJexlCalculatorResult(exp.get("expression").toString(), jexlContext);
            // 频率判断
            if (result && StringUtils.isNotBlank(policy.getCriterion())) {
                final Map<String, Object> criterion = JsonUtils.jsonToMap(policy.getCriterion());
                if (criterion.containsKey(DatabusPolicyConstant.FREQUENCY)) {
                    final String frequencyKey = PolicyUtil.getFrequencyKey(resourceBean.getCode(), policy);
                    if (!this.judgmentFrequency(frequencyKey,
                            JsonUtils.jsonToMap(criterion.get(DatabusPolicyConstant.FREQUENCY).toString()),
                            time.getTime())) {
                        return;
                    }
                }
            }
            eventBean.setExpressionResult(result);
            eventBean.setResourceBean(resourceBean);
            final List<Policy> policies = Lists.newArrayList();
            policies.add(policy);
            eventBean.setPolicies(policies);
            eventBean.setTime(time);
            this.sendResult(eventBean);
        } catch (final Exception e) {
            ComplexPolicyExpressionHandlerImpl.log.error("策略解析处理异常：{}", e);
            return;
        } finally {
            eventBean = null;
            resourceBean = null;
        }
    }

    /**
     * 触发频率判断方法
     *
     * @param similarid
     * @param frequency
     * @param time
     * @return
     * @throws POLCException
     */
    @SuppressWarnings("unchecked")
    protected boolean judgmentFrequency(final String frequencyKey, final Map<String, Object> frequency, final Long time)
            throws Exception {
        final String cacheName = DatabusPolicyConstant.POLICY_FREQUENCY_CACHE_NAME;
        if ((frequency.get(DatabusPolicyConstant.TIME) == null)
                || (frequency.get(DatabusPolicyConstant.TIMES) == null)) {
            throw new CustomRuntimeException(DatabusErrorCode.ERR_CODE_406220);
        }
        final Long timeFrequency = Long.parseLong(frequency.get(DatabusPolicyConstant.TIME).toString());
        final Integer timesFrequency = Integer.valueOf(frequency.get(DatabusPolicyConstant.TIMES).toString());

        if (timesFrequency < 2) {
            return true;
        }
        boolean result = false;
        List<Long> timeList = Lists.newLinkedList();
        if (this.cacheService.isInCache(cacheName, frequencyKey)) {
            timeList = (List<Long>) this.cacheService.get(cacheName, frequencyKey);
        }
        timeList.add(time);
        if ((time - timeList.get(0)) > timeFrequency) {
            timeList.remove(0);
            this.cacheService.put(cacheName, frequencyKey, timeList);
            result = false;
        }
        if (timeList.size() < timesFrequency) {
            this.cacheService.put(cacheName, frequencyKey, timeList);
            result = false;
        } else {
            this.cacheService.remove(cacheName, frequencyKey);
            result = true;
        }
        timeList = null;
        return result;
    }

    /**
     * 获取当前的指标采集数据
     *
     * @param complexPolicyKey
     * @param metrics
     * @param extendBean
     * @return
     */
    @SuppressWarnings("unchecked")
    private boolean getRealtime(final String complexPolicyKey, List<MetricBean> metrics,
            final ComplexPolicyExtendBean extendBean) {
        Map<String, Object> data = Maps.newHashMap();
        final String cacheName = DatabusPolicyConstant.POLICY_EXPRESSION_CACHE_NAME;
        final String metricCode = extendBean.getMetricCode();
        final Optional<MetricBean> optMetric =
                metrics.stream().filter(it -> it.getVariable().equals(metricCode)).findFirst();
        if (!optMetric.isPresent()) {
            return false;
        }
        try {
            if (this.cacheService.isInCache(cacheName, complexPolicyKey)) {
                data = (Map<String, Object>) this.cacheService.get(cacheName, complexPolicyKey);
            }
            data.put(extendBean.getExpressKey(), optMetric.get().getValue());
            this.cacheService.put(cacheName, complexPolicyKey, data);
            return true;
        } finally {
            metrics = null;
            data = null;
        }
    }

    /**
     * 获取历史指标采集数据
     *
     * @param complexPolicyKey
     * @param metrics
     * @param extendBean
     * @return
     */
    @SuppressWarnings("unchecked")
    private boolean getHistory(final String complexPolicyKey, List<MetricBean> metrics,
            final ComplexPolicyExtendBean extendBean) {
        Map<String, Object> data = Maps.newHashMap();
        List<MetricBean> hisDataList = Lists.newArrayList();
        final String cacheName = DatabusPolicyConstant.POLICY_EXPRESSION_CACHE_NAME;
        boolean result = false;
        final String metricCode = extendBean.getMetricCode();
        final Optional<MetricBean> optMetric =
                metrics.stream().filter(it -> it.getCode().equals(metricCode)).findFirst();
        if (!optMetric.isPresent()) {
            return false;
        }
        try {
            final String keyHistory = PolicyUtil.getHistoryKey(extendBean);
            if (this.cacheService.isInCache(cacheName, keyHistory)) {
                hisDataList = (List<MetricBean>) this.cacheService.get(cacheName, keyHistory);
                if (hisDataList.isEmpty()) {
                    result = false;
                } else {
                    hisDataList.sort((exp1, exp2) -> new Double(exp1.getCollectedTime().getTime())
                            .compareTo(new Double(exp2.getCollectedTime().getTime())));
                    if (this.cacheService.isInCache(cacheName, complexPolicyKey)) {
                        data = (Map<String, Object>) this.cacheService.get(cacheName, complexPolicyKey);
                    }
                    data.put(extendBean.getExpressKey(), hisDataList.get(0).getValue());
                    this.cacheService.put(cacheName, complexPolicyKey, data);
                    hisDataList.remove(0);
                    result = true;
                }
            }
            hisDataList.add(optMetric.get());
            this.cacheService.put(cacheName, keyHistory, hisDataList);
            return result;
        } finally {
            data = null;
            hisDataList = null;
            metrics = null;
        }
    }

    private boolean iSDataIntegrity(final List<ComplexPolicyExtendBean> complexPolicyExtendBeans,
            final JexlContext jexlContext) {
        for (final ComplexPolicyExtendBean extendBean : complexPolicyExtendBeans) {
            if (!jexlContext.has(extendBean.getExpressKey())) {
                return false;
            }
        }
        return true;
    }
}
